/*
 * This file is part of b.AMP.
 *
 * b.AMP is free software: you can redistribute it and/or modify it
 * under the terms of the GNU Lesser General Public License as
 * published by the Free Software Foundation, either version 3 of the
 * License, or (at your option) any later version.
 *
 * b.AMP is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with b.AMP.  If not, see <http://www.gnu.org/licenses/>.
 */

#include "message_ack.hpp"
#include "message_target.hpp"
#include "message_loop.hpp"

#include <bamp/message_ptr.hpp>

#include <tr1/functional>
#include <boost/thread.hpp>
#include <boost/interprocess/sync/interprocess_semaphore.hpp>
#include <boost/foreach.hpp>



/*
 * Public interface.
 */


bamp::MessageLoop::MessageLoop(const int nPriorities)
    : m_handlers      (),
      m_defaultHandler(),
      m_handlersMutex (),
      m_priorities    (),
      m_nPriorities   (nPriorities),
      m_prioFrozen    (new bool[nPriorities]),
      m_messages      (new MessageQueue[nPriorities]),
      m_messagesMutex (),
      m_messagesSem   (0),
      m_userMutex     ()
{
    for(int i = 0; i < nPriorities; ++i)
        m_prioFrozen[i] = false;
}


bamp::MessageLoop::~MessageLoop()
{
    // Stop the thread 
    stop();
    waitStop();
    
    // Delete all the copied messages.
    boost::mutex::scoped_lock messagesLock(m_messagesMutex);
    for(int i = 0; i < m_nPriorities; ++i) {
        while(m_messages[i].size() > 0) {
            m_messages[i].pop();
        }
    }

    delete[] m_messages;
    delete[] m_prioFrozen;
}


int bamp::MessageLoop::messagesPending()
{ 
    boost::mutex::scoped_lock lock(m_messagesMutex);
    int count = 0;
    for (int i = 0; i < m_nPriorities; ++i)
        count += m_messages[i].size();
    return count;
}


int bamp::MessageLoop::messagesPending(const int priority)
{ 
    boost::mutex::scoped_lock lock(m_messagesMutex);
    if (priority > m_nPriorities)
        return 0;
    else
        return m_messages[priority].size();
}


void bamp::MessageLoop::send(const Message& message, MessageAck ack)
{
    // If we have no handler for that message, do nothing.
    boost::mutex::scoped_lock handlersLock(m_handlersMutex);
    if ((m_handlers.find(message.getName()) == m_handlers.end()) &&
        !m_defaultHandler) {
        ack.ack();
        return;
    }
    handlersLock.unlock();

    // Create an internal copy of the message
    MessagePtr  copy = message.clone();
    MessagePair ep   = std::make_pair(copy, ack);

    // Find out the message's priority level
    boost::mutex::scoped_lock messagesLock(m_messagesMutex);
    PriorityMap::const_iterator it = m_priorities.find(message.getName());
    int priority = m_nPriorities-1;
    if (it != m_priorities.end())
        priority = it->second;

    // Add the message to the queue
    m_messages[priority].push(ep);
    messagesLock.unlock();

    // Notify the thread
    m_messagesSem.post();
}


bamp::MessageAck bamp::MessageLoop::send(const Message& message)
{
    MessageAck ack(1);
    send(message, ack);
    return ack;
}


void bamp::MessageLoop::connect(const std::string&     name,
                          const MessageFunction& func)
{
    boost::mutex::scoped_lock lock(m_handlersMutex);    
    m_handlers.insert(std::make_pair(name, func));
}


void bamp::MessageLoop::setDefault(const MessageFunction& func)
{
    boost::mutex::scoped_lock lock(m_handlersMutex);    
    m_defaultHandler = func;    
}


void bamp::MessageLoop::disconnect(const std::string& name)
{
    boost::mutex::scoped_lock lock(m_handlersMutex);    
    m_handlers.erase(name);
}


void bamp::MessageLoop::setPriority(const std::string& name, const int priority)
{
    boost::mutex::scoped_lock lock(m_messagesMutex);
    m_priorities.insert(std::make_pair(name, priority));
}


void bamp::MessageLoop::freezePriority(const int priority)
{
    boost::mutex::scoped_lock lock(m_messagesMutex);
    m_prioFrozen[priority] = true;
}


void bamp::MessageLoop::thawPriority(const int priority)
{
    boost::mutex::scoped_lock messagesLock(m_messagesMutex);
    for(size_t i = 0; i < m_messages[priority].size(); ++i)
        m_messagesSem.post();
    m_prioFrozen[priority] = false;
}


void bamp::MessageLoop::flushPriority(const int priority)
{
    boost::mutex::scoped_lock messagesLock(m_messagesMutex);
    while(!m_messages[priority].empty())
        m_messages[priority].pop();
}


boost::mutex::scoped_lock bamp::MessageLoop::lock()
{
    return boost::mutex::scoped_lock(m_userMutex);
}



/*
 * Private methods.
 */


bool bamp::MessageLoop::preRun()
{
    return true;
}


bool bamp::MessageLoop::postRun()
{
    return true;
}


bool bamp::MessageLoop::iteration()
{
    // Wait for an message
    m_messagesSem.wait();

    // Fetch the message with the lowest priority level
    boost::mutex::scoped_lock messagesLock(m_messagesMutex);
    int priority = -1;
    for (int i = 0; i < m_nPriorities; ++i) {
        if (!m_prioFrozen[i] && m_messages[i].size() > 0) {
            priority = i;
            break;
        }
    }

    if (priority < 0)
        return true;

    MessagePair ep      = m_messages[priority].front();
    MessagePtr  message = ep.first;
    MessageAck  ack     = ep.second;
    m_messages[priority].pop();
    messagesLock.unlock();

    // Execute the message's handlers. The handlers are copied to
    // a temporary vector to avoid calling them while m_handlers
    // is locked, which could create deadlocks. If there are no handlers,
    // the default handler is used, if any.
    std::string name = message.getName();
    std::vector<MessageFunction> handlers;

    boost::mutex::scoped_lock handlersLock(m_handlersMutex);
    HandlerMap::const_iterator it = m_handlers.find(name);

    for(; it != m_handlers.end() && it->first == name; ++it)
        handlers.push_back(it->second);

    if (handlers.empty() && m_defaultHandler) {
        handlers.push_back(m_defaultHandler);
    }

    handlersLock.unlock();

    {
        boost::mutex::scoped_lock userLock(m_userMutex);
        BOOST_FOREACH(MessageFunction f, handlers)
            f(message);
    }
        
    // Acknowkedge that the message has been handled by us
    ack.ack();

    return true;
}


bool bamp::MessageLoop::preStop()
{
    m_messagesSem.post();
    return true;
}


void bamp::MessageLoop::postStop()
{
    // Nothing here
}
